package com.qyer.log.job.sum;
import java.io.IOException;
import java.net.URISyntaxException;
import java.text.ParseException;
import java.text.SimpleDateFormat;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.qyer.log.job.sum.UserTracksSummary.Combiner;
import com.qyer.util.log.entity.TrackLog;
import com.qyer.util.log.service.LogService;
import com.qyer.util.log.service.impl.LogServiceImpl;
import com.qyer.util.log.util.LogFatory;
import com.qyer.util.log.util.MyDateUtil;

public class URLSummary {
    
	public static class M1 extends Mapper<LongWritable, Text, Text, Text> {
		 
		private Text val = new Text();
		private Text keyVal =new Text();
		
		final static String FS="\t";
		
		//清洗日志工具类
		LogService lli = (LogService)LogFatory.getInstance(LogServiceImpl.class.getName());
		
		TrackLog tracklog = null ;
		
		@Override
		public void map(LongWritable key, Text value, Context context) throws IOException,InterruptedException {
			
			tracklog= lli.getTrackFormat(value.toString());
				
			if(tracklog==null){
				return ; 
			}
			
			String becon_type=tracklog.getType();//获取becontype
			
			if("/open".equals(becon_type)){
				becon_type="1";
			}
			
			String url=tracklog.getCurUrl();//获取url地址
			String category=tracklog.getCategory();//获取唯一标示
			
			if(url.indexOf("?")>-1){
				url=url.split("\\?")[0];
			}
			
			if(url.length()>256||category.length()>128){
				return ; 
			}
			
			if(url.contains("bbs.qyer.com/thread")||url.contains("bbs.qyer.com//thread")){
				url="bbs.qyer.com/thread";
			}else if (url.contains("m.qyer.com/bbs/thread")||url.contains("m.qyer.com/bbs//thread")){
				url="m.qyer.com/bbs/thread";
			}else if (url.contains("ask.qyer.com/question")||url.contains("ask.qyer.com//question")){
				url="ask.qyer.com/question";
			}else if (url.contains("m.qyer.com/ask/question")||url.contains("m.qyer.com/ask//question")){
				url="m.qyer.com/ask/question";
			}
			
			url=url.replace("//", "/");
			
			if(url.endsWith("/")){
				url=url.substring(0 , url.length()-1);
			}
			
			String platform = tracklog.getPlatform(); 
			
			if("web".equals(platform)||"m".equals(platform)){
			
			    for (String u : new String []{"A" , url }){
			    	for (String c : new String []{"A" , category }){
			    		
			    		if(c.length()>128){
			    			continue;
			    		}
			    		
			    		if(u.length()>256){
			    			continue;
			    		}
		    			
		    			String keyStr = u+FS+c+FS+becon_type; 
		    			String valStr = "1";
		    			keyVal.set(keyStr);
		    			val.set(valStr);
		    			context.write(keyVal,val);    
				    }
			    }
				
			}	
		}         
	 }
	 
	 

	 
	public static class R1 extends Reducer<Text, Text, Text, Text> {
    
    	  private Text val = new Text();
 		  private Text keyVal =new Text();
 		  
 		  final static String FS="\t";
 		 final static String NEWFS="\001";
 		  
 		  static String DATE;
 		  
          @Override
          public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,InterruptedException {
        	if (DATE == null) {
  				DATE = context.getConfiguration().get("DATE");
  			}   
        	  
  			Long pv = 0l;
  			
  			if(key.toString().startsWith("A"+FS+"A")){
  				return ;
  			}

  			for (Text valT : values) {
  				String v =valT.toString();
				pv++;
  			}

  			String keyStr=key.toString().replace(FS, NEWFS);
  			keyVal.set(DATE+NEWFS+keyStr);
  			val.set(pv.toString());
  			context.write(keyVal, val);       
          }
         
     }
     /**
     * @param args
     * @throws URISyntaxException 
     * @throws ParseException 
     * @throws InterruptedException 
     * @throws ClassNotFoundException 
     */
     public static void main(String[] args) throws IOException, URISyntaxException, ParseException, ClassNotFoundException, InterruptedException{
    	 Configuration conf = new Configuration();
// 		conf.set("mapreduce.reduce.memory.mb" ,"3048");
 		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
 		int exitCode = 127;	 
			
		if (otherArgs.length !=1 ) {
			System.out.println("************************************************************");
			System.out.println("************************************************************");
			System.out.println("Usage: please input 1 params, for example: file.jar args[0] args[1]");
			System.out.println("args[0] is dateList: 20130101,20130106   or 20130106");
			System.out.println("************************************************************");
			System.out.println("************************************************************");
			System.exit(exitCode);
		}
          
		String startDate = otherArgs[0].split(",")[0];
		String endDate = otherArgs[0].split(",").length == 2 ? otherArgs[0].split(",")[1] : startDate;
		
		String baseWeb = "/flume/clean/track/";
		String baseOutPath = "/flume/summary/URLSummary";
		
		
		for (String runDate : MyDateUtil.getDateList(startDate, endDate, 0)) {
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
			Long startDateLong =sdf.parse(runDate).getTime();
			conf.set("DATE", startDateLong.toString());
			
			Job job = new Job(conf, "URLSummary");
			job.setInputFormatClass(TextInputFormat.class);
			job.setJarByClass(URLSummary.class);
			job.setMapperClass(M1.class);
			job.setReducerClass(R1.class);
			job.setNumReduceTasks(3);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(Text.class);
	        
	        
			for (String rDate : MyDateUtil.getDateList(startDate, endDate, 0)) {
 				System.out.println("runDate -------------->" +rDate);
 				FileInputFormat.addInputPath(job, new Path(baseWeb + rDate));
 			}
	        
	        String outPath = baseOutPath +"/"+ runDate;
	        FileSystem.get(conf).delete(new Path(outPath), true);
	        FileOutputFormat.setOutputPath(job, new Path(outPath));
	        
	        //启用压缩
	        FileOutputFormat.setCompressOutput(job, true);
	        //lzo格式
	        FileOutputFormat.setOutputCompressorClass(job, com.hadoop.compression.lzo.LzopCodec.class); 
	         
	        exitCode = job.waitForCompletion(true) ? 0 : 1;
	    	System.out.println("--------------------------------------------END1"+runDate);
		}
     }
}
